package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo03ReduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {
    /**
     * 微博热搜更新规则：每个10分钟，统计最近1小时的热度
     * 简化：每隔5s，统计最近15s单词数量
     * 上述需求实际上有两个时间因素：
     * 1、批次的大小（窗口的大小）
     * 2、计算任务的时间间隔
     * 需要使用时间窗口进行处理
     * 时间窗口主要有两种：
     * 1、滚动窗口：批次的大小=任务的时间间隔，窗口之间是独立的
     * 2、滑动窗口：批次的大小>任务的时间间隔，窗口之间有数据交叠的
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]")
      .getOrCreate()

    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    val textDS: DStream[String] = ssc.socketTextStream("master", 8888)
    textDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKeyAndWindow(_ + _, Durations.seconds(15)) // 指定窗口的大小为15s
    //      .print()


    /**
     * 例如：在第15s的时候，需要计算0-5s、5s-10s、10-15s的数据，合计：0-15s的数据
     * 那么在第20s的时候，需要计算5s-10s、10-15s、15-20s的数据，合计：5-20s的数据
     * 发现出现了重复计算
     * 优化：在第20s时的计算结果可以通过 第15s的结果减去(0-5s)结果再加上(15-20s)结果
     * 那么第5s-15s内的数据就不需要重复计算了
     * 需要设置ck的目录，因为需要保留之前的计算结果
     */

    ssc.checkpoint("spark/data/ck")
    textDS
      .flatMap(_.split(","))
      .map(word => (word, 1))
      .reduceByKeyAndWindow(_ + _, _ - _, Durations.seconds(15)) // 指定窗口的大小为15s
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
